package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import com.sun.xml.internal.bind.v2.TODO;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;

/**
 * @Description: 对用户行为数据进行分流, 消费Kafka中 ods_base_log 主题
 * @Author: tiancy
 * @Create: 2021/11/23
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关设置
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置检查点的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        //2.3 取消job是否保留检查点
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        //2.5 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/ck"));
        //2.6 设置操作Hadoop的默认用户为atguigu
        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 3.从Kafka中读取日志数据
        //3.1 声明消费的主题以及消费者组
        String topic = "ods_base_log";
        String groupId = "base_log_app_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        //TODO 4.对读取的数据 进行类型转换  jsonStr-->jsonObj
        /*
        //匿名内部类
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String jsonStr) throws Exception {
                return JSON.parseObject(jsonStr);
            }
        });

        //lambda表达式
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(
            jsonStr -> JSON.parseObject(jsonStr)
        );
        */
        // 对读取的数据 进行类型转换  json字符串-->json对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);

        // jsonObjDS.print(">>>>>");

        /*
            TODO 5.新老访客标记修复 ---> 状态编程
             需要先使用keyBy按照mid分组,使每个mid都存在键控状态,键控状态的类型为ValueState<String,存放的是上次访问时间>.再使用算子的状态对当前数据进行校验
                1、先判断当前json对象中`用户标识`是否为新用户,只有新用户才会判断,如果为老用户不用处理.
                2、判断当前用户是否存在设定的状态值,如果当前用户的状态值不是空的,则说明曾经访问过,
         */
        KeyedStream<JSONObject, String> jsonObjDSByMid = jsonObjDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        SingleOutputStreamOperator<JSONObject> jsonObjWithIsNewDS = jsonObjDSByMid
                .map(new RichMapFunction<JSONObject, JSONObject>() {  // 只是对带当前json对象中 is_new 进行操作,操作结果还需要将当前对象返回.
                    //声明状态  用于保存当前设备的上次访问日期
                    private ValueState<String> lastVisitDateState;
                    // 创建格式化时间的类
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastVisitDateState", Types.STRING));
                        sdf = new SimpleDateFormat("yyyyMMdd");
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObject) throws Exception {
                        String curMidHistoryState = lastVisitDateState.value();
                        //获取当前json对象的isNew状态 : 1为新用户. 可能由于程序宕机,造成数据丢情况,需要再次获取当前mid的is_New修复.
                        String isNewUser = jsonObject.getJSONObject("common").getString("is_new");
                        String currentVisitDate = sdf.format(jsonObject.getLong("ts"));
                        if ("1".equals(isNewUser)) {

                           /*
                                判断当前用户是否存在状态 :
                                    如果存在状态,则说明以前登录过,将is_New 字段置为0 .
                                    如果当前获取到的状态为空,则说明是首次访问,将首次访问时间设置到状态中.
                            */
                            if (StringUtils.isNotEmpty(curMidHistoryState)) {
                                // 曾经访问过,不是首次访问.
                                isNewUser = "0";  // 修改用户状态并更新到json对象中.

                                // jsonObject.getJSONObject("common").put("is_new", isNewUser);

                                // 这里还需要考虑一种情况,如果上次访问日期和当前访问日期为同一天,也就是一天内访问了多次,也算新用户.
                                if (!curMidHistoryState.equals(currentVisitDate))
                                    jsonObject.getJSONObject("common").put("is_new", isNewUser);
                            } else {
                                // 状态为空,首次访问.
                                lastVisitDateState.update(currentVisitDate);
                            }
                        }
                        return jsonObject;
                    }
                });

        /*
            TODO 6.日志数据进行分流       启动日志---启动侧输出流   曝光日志---曝光侧输出流中   页面日志---主流中
                侧输出流的作用主要有两个 : 1、处理迟到数据. 2、将一条分成多条流.
                使用步骤 :
                    1、先定义侧输出流的标签.是一个匿名类的方式 .
                    2、使用process算子,每来一条根据分流条件进行分流.并使用上下文对象 ctx.output(标签名称,写出的具体数据);
                    3、使用主流对象调用 getSideOutput(侧输出流标签) 获取到具体的侧输出流.
         */
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };

        //具体的分流操作,这里需要考虑,分流后,需要将数据写入到Kafka中,主题 dwd_start_log,dwd_display_log,dwd_page_log
        SingleOutputStreamOperator<String> pageDS = jsonObjWithIsNewDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<String> out) throws Exception {
                        //获取启动start标记
                        JSONObject startJsonObj = jsonObj.getJSONObject("start");
                        String jsonStr = jsonObj.toJSONString();

                        //判断是否为启动日志
                        if (startJsonObj != null && startJsonObj.size() > 0) {
                            //如果startJsonObj不为空，说明是启动日志  将当前日志输出到侧输出流中
                            ctx.output(startTag, jsonStr);
                        } else {
                            //如果不是启动日志，那么剩下的日志都属于页面日志  页面日志输出到主流中
                            out.collect(jsonStr);

                            //注意：曝光日志也属于页面日志     判断当前页面是否有曝光
                            JSONArray displayArr = jsonObj.getJSONArray("displays");
                            if (displayArr != null && displayArr.size() > 0) {
                                Long ts = jsonObj.getLong("ts");
                                String pageId = jsonObj.getJSONObject("page").getString("page_id");
                                //对数组进行遍历   将曝光数据一条条的取出来
                                for (int i = 0; i < displayArr.size(); i++) {
                                    JSONObject displayJsonObj = displayArr.getJSONObject(i);
                                    //补充曝光页面以及曝光的时间
                                    displayJsonObj.put("ts", ts);
                                    displayJsonObj.put("page_id", pageId);
                                    //将曝光信息输出曝光的侧输出流中
                                    ctx.output(displayTag, displayJsonObj.toJSONString());
                                }
                            }
                        }
                    }
                }
        );

        // 获取侧输出流
        DataStream<String> startSideOutput = pageDS.getSideOutput(startTag);
        DataStream<String> displaySideOutput = pageDS.getSideOutput(displayTag);
        startSideOutput.print("启动页面数据格式==>");
        displaySideOutput.print("曝光数据格式 ==>");
        pageDS.print("页面数据格式==>");

        //TODO 7.将不同流中的数据写到kafka的dwd主题中
        startSideOutput.addSink(MyKafkaUtil.getKafkaSink("dwd_start_log"));
        displaySideOutput.addSink(MyKafkaUtil.getKafkaSink("dwd_display_log"));
        pageDS.addSink(MyKafkaUtil.getKafkaSink("dwd_page_log"));
        env.execute();
    }
}
